package pers.cz.kettle;

import org.pentaho.di.core.exception.KettleStepException;
import org.pentaho.di.core.row.RowMeta;
import pers.cz.message.MessageType;
import pers.cz.message.MessageUtil;
import pers.cz.minio.MinioUtils;
import pers.cz.spring.SpringContextUtil;
import pers.cz.websocket.MessagePushSocketUtils;
import entity.DataBaseParamter;
import org.apache.commons.io.FileUtils;
import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.database.Database;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.exception.KettleXMLException;
import org.pentaho.di.core.plugins.PluginRegistry;
import org.pentaho.di.core.plugins.StepPluginType;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransHopMeta;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.steps.insertupdate.InsertUpdateMeta;
import org.pentaho.di.trans.steps.tableinput.TableInputMeta;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.security.core.context.SecurityContextHolder;

import java.io.File;
import java.io.IOException;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 数据同步工具
 * @program: PostGirl-panent
 * @description: DateSyncUtil
 * @author: Cheng Zhi
 * @create: 2021-05-17 15:44
 **/
public class DataSyncUtil {

    private static Logger log = LoggerFactory.getLogger(DataSyncUtil.class);

    /**
     * 初始化头文件
     * @param paramter 数据库连接信息
     * @return
     */
    public static String[] initDataBase(DataBaseParamter paramter) {

        String ip = paramter.getIp();
        String type = paramter.getType();
        String connectionName = paramter.getConnectionName();
        String dataBaseName = paramter.getDataBaseName();
        String port = paramter.getPort();
        String userName = paramter.getUserName();
        String password = paramter.getPassword();
        String[] dataBasesXML = {
                "<?xml version=\"1.0\" encoding=\"UTF-8\"?>" + "<connection>" + "<name>" + connectionName + "</name>"
                        + "<server>" + ip + "</server>" + "<type>" + type +"</type>" + "<access>Native</access>"
                        + "<database>" + dataBaseName + "</database>" + "<port>" + port + "</port>" + "<username>" + userName + "</username>"
                        + "<password>" + password + "</password>" + "</connection>"};


        return  dataBasesXML;
    }

    /**
     * 生成kettle文件,路径默认为./kettle/
     * @param tableNameA 输入表
     * @param tableNameB 输出表
     * @param sql 执行语句
     * @param primaryKey 表主键
     * @param updatelookup 要更新的字段
     * @return
     */
    public static TransMeta getKtr(DataBaseParamter paramterA, DataBaseParamter paramterB, String tableNameA, String tableNameB, String sql, String[] primaryKey, String[] updatelookup) throws KettleException, IOException {
        KettleEnvironment.init();
        TransMeta transMeta = generateTransform(paramterA, paramterB, tableNameA, tableNameB, sql, primaryKey, updatelookup);
        String transXml = "<?xml version=\"1.0\" encoding=\"UTF-8\"?> \n" + transMeta.getXML();
        String ktrPath = "./kettle/" + tableNameA + "-" + tableNameB + ".ktr";
        File file = new File(ktrPath);
        // 将kettle文件上传到文件存储服务器
        MinioUtils minioUtils = SpringContextUtil.getBean(MinioUtils.class);

        FileUtils.writeStringToFile(file, transXml, "UTF-8");
        log.debug("文件生成成功");
        try {
            minioUtils.makeBucket("post-girl"); // 创建存储桶
            String objectName = "Kettle/" + tableNameA + "-" + tableNameB + ".ktr";
            minioUtils.putObject("post-girl",objectName,ktrPath);
        } catch (Exception e) {
            log.error("文件上传服务器失败", e);
        }
        return null;
    }

    /**
     * 添加表输入
     * @param paramterA  source数据库信息
     * @param registry
     * @param transMeta
     * @param sql
     * @return
     */
    private static StepMeta addTableInput(DataBaseParamter paramterA, PluginRegistry registry, TransMeta transMeta, String sql) throws KettleStepException {

        TableInputMeta tableInput = new TableInputMeta();
        String tableInputPluginId = registry.getPluginId(StepPluginType.class, tableInput);
        // 给表输入添加一个DatabaseMeta连接数据库
        DatabaseMeta postGirlMysql = transMeta.findDatabase(paramterA.getConnectionName());
        tableInput.setDatabaseMeta(postGirlMysql);
        String selectSql = sql;
        tableInput.setSQL(selectSql);
        // 添加TableInputMeta到转换中
        StepMeta tableInputMetaStep = new StepMeta(tableInputPluginId, "table input", tableInput);
        // 给步骤添加在spoon工具中的显示位置
        tableInputMetaStep.setDraw(true);
        tableInputMetaStep.setLocation(100, 100);
        transMeta.addStep(tableInputMetaStep);
        return tableInputMetaStep;
    }

    /**
     * 添加表输出
     * @param registry
     * @param transMeta
     * @param tableName
     * @param primaryKey
     * @param updatelookup
     * @return
     */
    private static StepMeta addTableOutput(DataBaseParamter paramterB, PluginRegistry registry, TransMeta transMeta, String tableName, String[] primaryKey, String[] updatelookup) {

        InsertUpdateMeta insertUpdateMeta = new InsertUpdateMeta();
        String insertUpdateMetaPluginId = registry.getPluginId(StepPluginType.class, insertUpdateMeta);
        // 添加数据库连接
        DatabaseMeta postGirlOracle = transMeta.findDatabase(paramterB.getConnectionName());
        insertUpdateMeta.setDatabaseMeta(postGirlOracle);
        // 设置操作的表
        insertUpdateMeta.setTableName(tableName);
        insertUpdateMeta.setKeyLookup(primaryKey);
        insertUpdateMeta.setKeyStream(primaryKey);
        String[] keystream2 = new String[primaryKey.length];
        String[] keyCondition2 = new String[primaryKey.length];
        for (int i=0; i<primaryKey.length; i++) {
            keystream2[i] = "";
            keyCondition2[i] = "=";
        }
        insertUpdateMeta.setKeyStream2(keystream2);
        insertUpdateMeta.setKeyCondition(keyCondition2);
        // 设置要更新的字段
        insertUpdateMeta.setUpdateLookup(updatelookup);
        insertUpdateMeta.setUpdateStream(updatelookup);

        Boolean[] updateOrNot = new Boolean[updatelookup.length];
        for (int i=0; i<updatelookup.length; i++) {
            updateOrNot[i] = true;
        }
        insertUpdateMeta.setUpdate(updateOrNot);

        // 添加步骤到转换中
        StepMeta insertUpdateStep = new StepMeta(insertUpdateMetaPluginId, "insert_update", insertUpdateMeta);
        insertUpdateStep.setDraw(true);
        insertUpdateStep.setLocation(250, 100);
        transMeta.addStep(insertUpdateStep);

        return insertUpdateStep;
    }

    /**
     * 添加转换
     * @param paramterA
     * @param paramterB
     * @param tableNameA
     * @param tableNameB
     * @param sql
     * @param primaryKey
     * @param updatelookup
     * @return
     */
    private static TransMeta generateTransform(DataBaseParamter paramterA, DataBaseParamter paramterB, String tableNameA, String tableNameB, String sql, String[] primaryKey, String[] updatelookup) throws KettleException {

        String[] dataBasesXmlA = initDataBase(paramterA);
        // 数组扩容并添加数据库连接
        String[] dataBaseXML = new String[dataBasesXmlA.length * 2];
        dataBaseXML[0] = dataBasesXmlA[0];
        dataBaseXML[1] = initDataBase(paramterB)[0];

        TransMeta transMeta = new TransMeta();
        // 设置转化元名称
        transMeta.setName("PostGirl-DataSync");
        // 添加转化数据库连接
        for (int i=0; i<dataBaseXML.length; i++) {
            try {
                DatabaseMeta databaseMeta = new DatabaseMeta(dataBaseXML[i]);
                transMeta.addDatabase(databaseMeta);
            } catch (KettleXMLException e) {
                log.error("添加添加转化数据库连接xml异常", e);
            }
        }
        // registry是给每个步骤生成一个标识Id用
        PluginRegistry registry = PluginRegistry.getInstance();
        StepMeta tableInputMetaStep = addTableInput(paramterA, registry, transMeta, sql);
        if (updatelookup == null || updatelookup.length == 0) {
            // 如果没有输入更新字段则，默认更新所有字段
            updatelookup = getInputFields(paramterA, registry, transMeta,sql);
        }
        StepMeta insertUpdateStep = addTableOutput(paramterB, registry, transMeta, tableNameB, primaryKey, updatelookup);
        transMeta.addTransHop(new TransHopMeta(tableInputMetaStep, insertUpdateStep));
        return transMeta;
    }

    /**
     * 获取源头表中的字段
     * @param paramterA
     * @param registry
     * @param transMeta
     * @return
     * @throws KettleStepException
     */
    public static String[] getInputFields(DataBaseParamter paramterA, PluginRegistry registry,TransMeta transMeta,String sql) throws KettleStepException {
        TableInputMeta tableInput = new TableInputMeta();
        String tableInputPluginId = registry.getPluginId(StepPluginType.class, tableInput);
        // 给表输入添加一个DatabaseMeta连接数据库
        DatabaseMeta postGirlMysql = transMeta.findDatabase(paramterA.getConnectionName());
        tableInput.setDatabaseMeta(postGirlMysql);
        tableInput.setSQL(sql);
        RowMeta rowMeta = new RowMeta();
        tableInput.getFields(rowMeta,null, null, null, null, null, null);
        String[] fieldNames = rowMeta.getFieldNames();

        return fieldNames;
    }
    /**
     * 执行转换
     * @param paramterA
     * @param paramterB
     * @param tableNameA
     * @param tableNameB
     * @param sql
     * @param primaryKey
     * @param updatelookup
     * @throws KettleException
     */
    public static void executeTransForm(DataBaseParamter paramterA, DataBaseParamter paramterB, String tableNameA, String tableNameB, String sql, String[] primaryKey, String[] updatelookup) throws KettleException {

        KettleEnvironment.init();
        TransMeta transMeta = generateTransform(paramterA, paramterB, tableNameA, tableNameB, sql, primaryKey, updatelookup);
        Trans trans = new Trans(transMeta);
        trans.execute(null); // You can pass arguments instead of null.
        // 等待执行完成
        trans.waitUntilFinished();

        String content = "";
        if (trans.getErrors()!=0) {
            log.debug("同步失败！");
            content = "notify|数据同步失败！";
        } else {
            log.debug("同步成功！");
            content = "notify|数据同步成功！";
        }

        // 将结果记录到数据库。默认都是未读的
        MessageUtil.insertNotifyMessage(content, MessageType.getName("notify"), "未读", null);

        // 发送实时消息
        MessagePushSocketUtils.sendInfo(content,Integer.valueOf(SecurityContextHolder.getContext().getAuthentication().getPrincipal().toString()));
    }

    /**
     * 执行sql
     * @param paramter 数据源信息
     * @param sql
     * @return
     * @throws KettleException
     * @throws SQLException
     */
    public static List<String> executeSql(DataBaseParamter paramter, String sql) throws KettleException, SQLException {

        // 初始化数据库
        KettleEnvironment.init();
        // 初始化数据库连接
        String[] dataBasesXml = initDataBase(paramter);
        // 创建转换
        TransMeta transMeta = new TransMeta();
        // 设置转化元名称
        transMeta.setName("PostGirl-DataSync");
        // 转化元添加数据库连接
        for (int i=0; i<dataBasesXml.length; i++) {
            try {
                DatabaseMeta databaseMeta = new DatabaseMeta(dataBasesXml[i]);
                transMeta.addDatabase(databaseMeta);
            } catch (KettleXMLException e) {
                log.error("添加添加转化数据库连接xml异常", e);
            }
        }
        Database targetDatabase = new Database(transMeta.findDatabase(paramter.getConnectionName()));
        targetDatabase.connect();
        ResultSet resultSet = targetDatabase.openQuery(sql);
        int col = resultSet.getMetaData().getColumnCount();
        List<String> list = new ArrayList<String>();
        while (resultSet.next()) {
            for (int i = 1; i <= col; i++) {
                list.add(resultSet.getString(i));
            }
        }
        return list;
    }

    public static List<Map<String, Object>> executeSqlToMap(DataBaseParamter paramter, String sql) throws KettleException {
        // 初始化数据库
        KettleEnvironment.init();
        // 初始化数据库连接
        String[] dataBasesXml = initDataBase(paramter);
        // 创建转换
        TransMeta transMeta = new TransMeta();
        // 设置转化元名称
        transMeta.setName("PostGirl-DataSync");
        // 转化元添加数据库连接
        for (int i=0; i<dataBasesXml.length; i++) {
            try {
                DatabaseMeta databaseMeta = new DatabaseMeta(dataBasesXml[i]);
                transMeta.addDatabase(databaseMeta);
            } catch (KettleXMLException e) {
                log.error("添加添加转化数据库连接xml异常", e);
            }
        }
        Database targetDatabase = new Database(transMeta.findDatabase(paramter.getConnectionName()));
        targetDatabase.connect();
        ResultSet resultSet = targetDatabase.openQuery(sql);

        return dealMapResultSet(resultSet);
    }

    /**
     * 将reslutSet映射为Map
     * @param resultSet
     * @return
     */
    private static List<Map<String, Object>> dealMapResultSet(ResultSet resultSet) {

        try {
            // 获取数据库字段信息
            ResultSetMetaData metaData = resultSet.getMetaData();
            int columnCount = metaData.getColumnCount();
            List<String> columnNameList = new ArrayList<>();
            for (int i = 1; i<= columnCount; i++) {
                columnNameList.add(metaData.getColumnName(i));
            }
            List<Map<String, Object>> mapList = new ArrayList<>();
            while (resultSet.next()) {
                Map<String, Object> map = new HashMap<String, Object>();
                for (String columnName: columnNameList) {
                    map.put(columnName.toUpperCase(), resultSet.getObject(columnName));
                }
                mapList.add(map);
            }

            return mapList;
        } catch (SQLException e) {
            e.printStackTrace();
        }

        return null;
    }
}
